package demo_day4;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;

public class MyProduce {
    // 需求：要生成一个kafka的客户端，完成数据的生产
    public static void main(String[] args) throws InterruptedException {
        // 1、创建一个map，存放各种配置信息（连接到哪个kafka集群等等）
        // MAP的泛型，不能直接 String String, 因为有的value 是一个class的映射
        HashMap<String,Object> confMap = new HashMap<>();
        // 添加各种配置
        confMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop10:9092");
        // 这里的key 是分区的时候用的
        confMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        confMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        confMap.put(ProducerConfig.LINGER_MS_CONFIG,0); // 等待5秒钟 不设置的话默认是0毫秒
        confMap.put(ProducerConfig.BATCH_SIZE_CONFIG,60); // 缓存区的大小
        // 如果你编写了一个拦截器，那么你需要map配置对象添加它
        confMap.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"demo_day4.KafkaIntercept");

        // 2、创建一个producer客户端，接入配置map对象，完成客户端 创建
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(confMap);

        // 3、基于客户端，完成数据的生产
//        ProducerRecord<String,String> record = new ProducerRecord<>("topicB","from_java");
//        producer.send(record); // 发送数据
        for (int i = 0; i <= 10; i++) {
            producer.send(new ProducerRecord<>("topicDD","hello"+i +","+ (10000+i*100)));
        }

        Thread.sleep(10000);

        // 4、关闭资源,会直接将缓冲区全部清空
        producer.close();
    }
}
